package com.maverick.util;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class KafkaUtil {
    private static final String DEFAULT_TOPIC = "default_topic";

    public static FlinkKafkaConsumer<String> getKafkaSource(String topic, String groupId, String kafkaServer) {
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        return new FlinkKafkaConsumer<>(topic, new SimpleStringSchema(), props);
    }

    public static FlinkKafkaProducer<String> getKafkaSink(String topic, String kafkaServer) {
        Properties props = new Properties();
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
        props.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 15 * 60 * 1000L + "");
        return new FlinkKafkaProducer<>(
                DEFAULT_TOPIC,
                (KafkaSerializationSchema<String>) (str, timestamp) -> new ProducerRecord<>(topic, str.getBytes()),
                props,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
    }

    public static <T> FlinkKafkaProducer<T> getKafkaSinkBySchema(
            KafkaSerializationSchema<T> kafkaSerializationSchema, String kafkaServer) {
        Properties props = new Properties();
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
        props.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 15 * 60 * 1000L + "");
        return new FlinkKafkaProducer<>(
                DEFAULT_TOPIC,
                kafkaSerializationSchema,
                props,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
    }

    public static DataStreamSource<String> getKafkaSourceStream(StreamExecutionEnvironment env,
                                                                String topic,
                                                                String groupId,
                                                                String kafkaServer) {
        FlinkKafkaConsumer<String> kafkaSrc = getKafkaSource(topic, groupId, kafkaServer);
        return env.addSource(kafkaSrc);
    }

    //拼接Kafka相关属性到DDL
    public static String getKafkaDDL(String topic, String groupId, String kafkaServer) {
        return  "'connector' = 'kafka', " +
                "'topic' = '" + topic + "', " +
                "'properties.bootstrap.servers' = '" + kafkaServer + "', " +
                "'properties.group.id' = '" + groupId + "', " +
                "'format' = 'json', " +
                "'scan.startup.mode' = 'latest-offset'";
    }
}

